package cn.admin.remoting.serialize;

import cn.admin.remoting.Remoting;
import cn.admin.remoting.serialize.impl.SliceOutputStream;
import com.caucho.hessian.io.HessianInput;
import com.caucho.hessian.io.HessianOutput;

import java.io.*;
import java.util.WeakHashMap;

public class DefaultSerialization implements CustomSerialization<Object> {

    public static final DefaultSerialization instance = new DefaultSerialization();

    @Override
    public void serialize(Object msg, byte protocol, OutputStream out)
            throws IOException {

        switch (protocol) {
            case 1:
                createHessianOutput(out).writeObject(msg);
                return;
            case 2:
            case 3:
                ObjectOutputStream javaOutput;
                javaOutput = Remoting.serialUtil().isReuseSerialization() ?
                        new JavaObjectOutputStream(out) :
                        new ObjectOutputStream(out);
                try {
                    javaOutput.writeObject(msg);
                    javaOutput.flush();
                } finally {
                    javaOutput.close();
                }
                return;
            default:
                throw new IllegalArgumentException("不可识别的序列化协议:" + protocol);
        }

    }

    private static HessianOutput createHessianOutput(OutputStream out) {
        HessianOutput hout = Remoting.serialUtil().isReuseSerialization() ?
                new HessianObjectOutputStream(out) : new HessianOutput(out);
        hout.setSerializerFactory(Remoting.serialUtil().getHessianSerializerFactory());
        return hout;
    }

    private static HessianInput createHessianInput(InputStream in) {
        ClassLoader tccl = Thread.currentThread().getContextClassLoader();
        if (tccl == null) {
            Thread.currentThread().setContextClassLoader(
                    DefaultSerialization.class.getClassLoader());
        }
        HessianInput hin = new HessianInput(in);
        hin.setSerializerFactory(Remoting.serialUtil().getHessianSerializerFactory());
        return hin;
    }

    @Override
    public Object deserialize(InputStream input, byte protocol)
            throws IOException {

        switch (protocol) {
            case 1:
                try {
                    HessianInput hin = createHessianInput(input);
                    return hin.readObject();
                } catch (IOException e) {
                    IOException newIoe =
                            new IOException(e.getMessage() + ", " +
                                    HessianInput.class.getResource("HessianInput.class"));
                    newIoe.setStackTrace(e.getStackTrace());
                    throw newIoe;
                }
            case 2:
            case 3:
                ObjectInputStream javaInput = new ObjectInputStream(input);
                Object obj;
                try {
                    obj = javaInput.readObject();
                } catch (ClassNotFoundException e) {
                    throw new IOException("ClassNotFoundException:" + e.getMessage(),
                            e.getCause());
                } finally {
                    javaInput.close();
                }
                return obj;
            default:
                throw new IllegalArgumentException("不可识别的序列化协议:" + protocol);
        }
    }

    @SuppressWarnings("unchecked")
    public <T> T hessianReadObject(InputStream is,Class<T> type) throws IOException {

        HessianInput hin = createHessianInput(is);
        try {
            return (T) hin.readObject(type);
        } catch (IOException e) {
            IOException newIoe =
                    new IOException(e.getMessage() + ", "
                            + HessianInput.class.getResource("HessianInput.class"));
            newIoe.setStackTrace(e.getStackTrace());
            throw newIoe;
        }

    }

    private static class HessianObjectOutputStream extends HessianOutput {

        private static WeakHashMap<SerialDataReusable,byte[]> reuseSerialCache =
                new WeakHashMap<>();

        HessianObjectOutputStream(OutputStream os) {
            super(os);
        }

        @Override
        public void writeObject(Object object) throws IOException {
            if (!(object instanceof SerialDataReusable)) {
                super.writeObject(object);
            } else {
                SerialDataReusable reuseSerialObj = (SerialDataReusable) object;
                byte[] reuseSerial = reuseSerialCache.get(reuseSerialObj);
                if (reuseSerial == null) {
                    ByteArrayOutputStream binary = new ByteArrayOutputStream();
                    HessianOutput hout = new HessianOutput(binary);
                    hout.setSerializerFactory(Remoting.serialUtil().getHessianSerializerFactory());
                    hout.writeObject(object);
                    reuseSerial = binary.toByteArray();
                    reuseSerialCache.put(reuseSerialObj,reuseSerial);
                }

                if (this.os instanceof SliceOutputStream) {
                    ((SliceOutputStream)this.os).addSlice(reuseSerial);
                } else {
                    os.write(reuseSerial);
                }
            }
        }
    }

    private static class JavaObjectOutputStream extends ObjectOutputStream {

        private static WeakHashMap<SerialDataReusable,byte[]> reuseSerialCache =
                new WeakHashMap<>();

        private final OutputStream os;
        private final ObjectOutputStream delegate;

        JavaObjectOutputStream(OutputStream out) throws IOException {
            this.os = out;
            this.delegate = new ObjectOutputStream(out);
        }

        @Override
        public void flush() throws IOException {
            this.delegate.flush();
        }

        @Override
        public void close() throws IOException {
            this.delegate.close();
        }

        @Override
        protected void writeObjectOverride(Object obj) throws IOException {
            if (!(obj instanceof SerialDataReusable)) {
                this.delegate.writeObject(obj);
            } else {
                SerialDataReusable reuseSerialObj = (SerialDataReusable) obj;
                byte[] reuseSerial = reuseSerialCache.get(reuseSerialObj);
                this.delegate.flush();
                if (reuseSerial == null) {
                    ByteArrayOutputStream binary = new ByteArrayOutputStream();
                    ObjectOutputStream oos = new ObjectOutputStream() {
                        @Override
                        protected void writeStreamHeader() throws IOException {
                        }
                    };
                    oos.writeObject(reuseSerialObj);
                    oos.flush();
                    oos.close();
                    reuseSerial = binary.toByteArray();
                    reuseSerialCache.put(reuseSerialObj,reuseSerial);
                }

                if (this.os instanceof SliceOutputStream) {
                    ((SliceOutputStream)this.os).addSlice(reuseSerial);
                } else {
                    this.os.write(reuseSerial);
                }
            }
        }
    }

}
